package com.shujia.ml

import org.apache.spark.ml.classification.{LogisticRegression, LogisticRegressionModel}
import org.apache.spark.ml.feature.LabeledPoint
import org.apache.spark.ml.linalg.Vectors
import org.apache.spark.sql.{DataFrame, SparkSession}

object Code06ModelLoad {
  def main(args: Array[String]): Unit = {
    val spark: SparkSession = SparkSession
      .builder()
      .master("local[*]")
      .appName("spark")
      .getOrCreate()

    import spark.implicits._

    // 加载模型，该路径中保存了模型中的各种参数以及未知变量的数据
    val logisticRegression: LogisticRegressionModel = LogisticRegressionModel.load("spark_code/model/logistic")


    val dataFrame: DataFrame = spark.read.format("image").load("spark_code/data/ml/logistic_jpg")
    dataFrame.printSchema()

    val featuresDF: DataFrame = dataFrame
      .select($"image.origin" as "path", $"image.data" as "data")
      .rdd
      .map {
        case row => {
          val path: String = row.getAs[String]("path")
          val pointArr = row.getAs[Array[Byte]]("data")
          val pointList: List[Double] = pointArr.toList.map {
            case point => if (point >= 0) {
              255.0
            } else {
              0.0
            }
          }
          ( path,Vectors.dense(pointList.toArray))
        }
      }.toDF("path","features")


    val res: DataFrame = logisticRegression.transform(featuresDF)
    res.show(truncate = false)

    //    logisticRegression.transform()

  }
}
